package com.hup.utils.io.netty;

import com.hup.utils.commons.exception.AppExceptionHandler;
import com.hup.utils.io.netty.NettyTCPClient.ClientConfig;
import com.hup.utils.io.netty.NettyUtil.BindCallback;
import com.hup.utils.io.netty.handler.*;
import com.hup.utils.io.netty.handler.DataHandler.DataMethodInvoker;
import com.hup.utils.io.netty.handler.HeartbeatHandler.IdleCallback;
import com.hup.utils.io.netty.model.BaseDataPacket;
import com.hup.utils.io.netty.model.LoginRequestPacket;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.log4j.Log4j2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * netty,nio,TCP服务端;<pre>
 * 注意:服务端的线程池在
 * .{@link #stopServer()}时就会完全关闭
 * .在 {@link #startServer()}时会根据配置重新构建,因为服务端一般很少频繁启动
 *
 * @author hugan
 * @date 2021/4/24
 */
@Log4j2
public class NettyTCPServer {
    private final ServerConfig serverConfig;
    private final StartedObj startedObj;
    private final ServerStateHandler serverStateHandlerInstance;
    private final ClientHandler clientHandlerInstance;

    public NettyTCPServer(ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
        startedObj = new StartedObj();
        startedObj.dataHandlerMap = serverConfig.registerDataHandlerMap();
        clientHandlerInstance = new ClientHandler();
        serverStateHandlerInstance = new ServerStateHandler();
    }

    /**
     * 启动服务端, 启动配置为{@link ServerConfig}
     */
    public void startServer() {
        startedObj.assertState(StartState.STOPPED);
        log.info("正在启动服务端");
        startedObj.startState = StartState.STARTING;

        ServerBootstrap bootstrap = new ServerBootstrap();
        startedObj.buildExecutor(serverConfig);
        bootstrap.group(startedObj.bossGroup, startedObj.workerGroup)
                .channel(NioServerSocketChannel.class)
                .handler(serverStateHandlerInstance)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel c) {
                        //构建pipeline处理连,每个客户端连接时被调用
                        log.info("客户端连接中={}", c.remoteAddress());
                        c.pipeline().addLast(new DataSplitter());
                        c.pipeline().addLast(DataCodecHandler.INSTANCE);
                        if (serverConfig.loginHandler != null) c.pipeline().addLast(serverConfig.loginHandler);
                        if (serverConfig.addHeartbeat) HeartbeatHandler.addHeartbeatHandler(c.pipeline());
                        c.pipeline().addLast(clientHandlerInstance);
                    }
                });
        //log.info("dataHandlerMap={}", TextUtil.getLineWrapLog(startedObj.dataHandlerMap.entrySet()));
        NettyUtil.bindPort(bootstrap, serverConfig.originalPort, serverConfig.bindRetryTime, serverStateHandlerInstance);
    }

    /**
     * 关闭服务端
     */
    public void stopServer() {
        startedObj.assertState(StartState.STARTED);
        log.info("正在关闭服务端");
        startedObj.serverContext.close().addListener(future -> serverStateHandlerInstance.onServerStopResult(future.isSuccess()));
    }

    public int getBoundPort() {
        startedObj.assertState(StartState.STARTED);
        return startedObj.boundPort;
    }

    public StartState getStartState() {
        return startedObj.startState;
    }

    /**
     * 用于监听服务端启动关闭的生命周期
     * bootstrap.handler(serverStateHandlerInstance)后的被调用过程
     * . {@link #startServer()}
     * . {@link ServerStateHandler#channelUnregistered(ChannelHandlerContext)}
     * . {@link ServerStateHandler#handlerRemoved(ChannelHandlerContext)}
     * . {@link ServerStateHandler#channelActive(ChannelHandlerContext)}
     * 所以 {@link #stopServer()}时不能通过 ChannelInboundHandlerAdapter的原生api监听服务关闭结果!
     */
    @Sharable
    private class ServerStateHandler extends ChannelInboundHandlerAdapter implements BindCallback {
        @Override
        public void onBindResult(boolean success, Integer boundPort) {
            log.info("success={}, boundPort={}", success, boundPort);
            startedObj.boundPort = success ? boundPort : null;
            startedObj.startState = success ? StartState.STARTED : StartState.STOPPED;
            serverConfig.serverCallback.onServerStartResult(success, boundPort);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.info("{}", ctx.channel());
            startedObj.serverContext = ctx;//获取ctx用于主动关闭服务
            super.channelActive(ctx);
        }

        private void onServerStopResult(boolean success) {
            if (!success) {
                log.error("服务端关闭失败??");
            } else {
                log.info("服务端已关闭");
                startedObj.startState = StartState.STOPPED;
                startedObj.releaseExecutor();
                startedObj.serverContext = null;
                serverConfig.serverCallback.onServerStopped();
            }
        }
    }

    /**
     * 客户端处理器
     */
    @Sharable
    private class ClientHandler extends SimpleChannelInboundHandler<BaseDataPacket> {

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            if (serverConfig.loginHandler == null) {
                //无登陆认证功能时,此处代表连接成功,马上通知callback
                Channel channel = ctx.channel();
                log.info("客户端连接成功={}", channel.remoteAddress());
                serverConfig.serverCallback.onClientConnectChanged(true, channel);
            } else {
                ctx.channel().eventLoop().schedule(() -> {
                    Channel channel = ctx.channel();
                    if (!LoginHandler.isLoginValidSuccess(channel)) {
                        log.warn("客户端登陆认证失败/无认证,主动断开连接, {}", channel.remoteAddress());
                        channel.close();
                    }
                }, serverConfig.getLoginValidTimeout(), TimeUnit.MILLISECONDS);
            }
            super.channelActive(ctx);
        }

        /**
         * 处理socket收到的数据
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, BaseDataPacket msg) {
            if (serverConfig.loginHandler != null && msg instanceof LoginRequestPacket) {
                /*
                有登陆认证功能且认证已通过,此处只能是由LoginHandler传递过来的,肯定不会是不符合配置的 LoginRequestPacket实现类
                则通知callback
                 */
                serverConfig.serverCallback.onClientConnectChanged(true, ctx.channel());
                return;
            }
            DataMethodInvoker invoker = startedObj.dataHandlerMap.get(msg.getClass());
            if (invoker != null) {
                //已指定处理方法的数据:通过ServerConfig.dataHandler配置指定的
                try {
                    //log.info("mgs={}", msg);
                    invoker.invoke(msg, ctx.channel());
                } catch (Exception e) {
                    log.error("消息处理异常={}", msg);
                    AppExceptionHandler.handle("消息处理异常", e);
                }
            } else {
                //未知数据/未指定处理方法的数据
                serverConfig.serverCallback.onUnknownData(ctx, msg);
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            if (serverConfig.loginHandler == null || LoginHandler.isLoginValidSuccess(channel)) {
                //没有登陆认证功能,或已经通过认证,则断线后通知callback;
                log.info("客户端连接断开={}", channel.remoteAddress());
                serverConfig.serverCallback.onClientConnectChanged(false, channel);
            } else {
                // else 不需要通知callback!
                log.warn("未通过认证的连接断开={}", channel.remoteAddress());
            }
            super.channelInactive(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
            serverConfig.serverCallback.serverExceptionCaught(ctx, e);
        }
    }

    /**
     * 服务端启动后的状态对象
     */
    private static class StartedObj {
        private static final Map<Class<? extends BaseDataPacket>, DataMethodInvoker> EMPTY_METHOD_MAP = new HashMap<>();
        /**
         * 服务端启动状态
         */
        private StartState startState = StartState.STOPPED;
        private NioEventLoopGroup bossGroup;
        private NioEventLoopGroup workerGroup;
        /**
         * 实际已绑定的端口号
         */
        private Integer boundPort;
        /**
         * 服务端context,用于主动关闭服务
         */
        private ChannelHandlerContext serverContext;
        private Map<Class<? extends BaseDataPacket>, DataMethodInvoker> dataHandlerMap;

        private void buildExecutor(ServerConfig serverConfig) {
            if (serverConfig.nioEventLoopGroupBuilder != null) {
                bossGroup = serverConfig.nioEventLoopGroupBuilder.get(false);
                workerGroup = serverConfig.nioEventLoopGroupBuilder.get(true);
            } else {
                bossGroup = new NioEventLoopGroup();
                workerGroup = new NioEventLoopGroup();
            }
        }

        private void releaseExecutor() {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            bossGroup = null;
            workerGroup = null;
        }

        private void assertState(StartState state) {
            if (startState != state) throw new IllegalArgumentException("服务状态为:" + startState.desc);
        }
    }

    /**
     * 服务端启动配置类;
     * 配置在下一次 {@link NettyTCPServer#startServer()}时生效
     */
    @SuppressWarnings("UnusedReturnValue")
    public static class ServerConfig implements DataHandlerRegister {
        private int originalPort;
        private int bindRetryTime = 5;
        private ServerCallback serverCallback;
        @Getter
        private LoginHandler<?, ?> loginHandler;
        /**
         * 客户端连接成功后,n毫秒内没认证通过,则主动断开连接;
         * 有登陆认证时业务时才使用
         */
        @Getter
        @Setter
        private int loginValidTimeout = 1000;
        private boolean addHeartbeat;
        /**
         * 数据处理类,每个元素相当于一个Controller
         */
        @Getter
        private ArrayList<DataHandler> lstDataHandler = new ArrayList<>();
        private NioEventLoopGroupBuilder nioEventLoopGroupBuilder;

        /**
         * @param serverCallback notNull
         */
        public ServerConfig(int port, ServerCallback serverCallback) {
            originalPort = port;
            if (serverCallback == null) throw new IllegalArgumentException("serverCallback不能为空");
            this.serverCallback = serverCallback;
        }

        public ServerConfig port(int port) {
            originalPort = port;
            return this;
        }

        /**
         * {@link NettyUtil#bindPort(ServerBootstrap, int, int, BindCallback)}的bindRetryTime
         */
        public ServerConfig bindRetryTime(int bindTryTime) {
            this.bindRetryTime = bindTryTime;
            return this;
        }

        /**
         * 配置登陆认证功能;
         * [登陆认证功能]双向配置流程:
         * .服务端初始化时,调用本方法
         * .客户端调用 {@link ClientConfig#loginHandler(LoginResponseHandler)}
         *
         * @param loginHandler null-移除登录认证功能;notNull-添加登陆认证功能
         */
        public ServerConfig loginHandler(LoginHandler<?, ?> loginHandler) {
            this.loginHandler = loginHandler;
            LoginHandler.handleLoginRequestCodec(loginHandler);
            return this;
        }

        /**
         * 配置有无心态机制; {@link HeartbeatHandler}
         */
        public ServerConfig setAddHeartbeat(boolean addHeartbeat, IdleCallback idleCallback) {
            return setAddHeartbeat(addHeartbeat, idleCallback, HeartbeatHandler.getReaderIdleTimeSeconds());
        }

        /**
         * @param readerIdleTimeSeconds {@link HeartbeatHandler#getReaderIdleTimeSeconds()}
         */
        public ServerConfig setAddHeartbeat(boolean addHeartbeat, IdleCallback idleCallback, int readerIdleTimeSeconds) {
            this.addHeartbeat = addHeartbeat;
            if (addHeartbeat && idleCallback == null) throw new IllegalArgumentException("idleCallback不能为空");
            HeartbeatHandler.initHeartBeat(false, addHeartbeat ? idleCallback : null, readerIdleTimeSeconds);
            return this;
        }

        /**
         * 配置自定义的线程池, 注意: {@link NettyTCPServer}
         */
        public ServerConfig nioEventLoopGroupBuilder(NioEventLoopGroupBuilder nioEventLoopGroupBuilder) {
            this.nioEventLoopGroupBuilder = nioEventLoopGroupBuilder;
            return this;
        }

    }

    /**
     * 服务端启动状态
     */
    public enum StartState {
        STOPPED("已关闭"),
        STARTED("已启动"),
        STARTING("启动中");
        private final String desc;

        StartState(String desc) {
            this.desc = desc;
        }
    }

    /**
     * 服务端状态回调
     */
    public interface ServerCallback {

        void onServerStartResult(boolean success, Integer boundPort);

        void onServerStopped();

        /**
         * 客户端上下线时的回调方法,可以在此处理SessionList
         *
         * @param onLine true-客户端连接(且认证)成功;false-客户端连接断开
         * @apiNote 注意:此处更新Session时,由于用户同时上下线时会有n个不同线程调用此方法,所以会有(List,Map)线程安全问题,所以应该加入synchronized;<br>
         * 并且服务器关闭时,建议在 {@link #onServerStopped()}中统一处理客户端下线的Session业务,而不是频繁地在此方法逐个处理session并通知ui
         */
        void onClientConnectChanged(boolean onLine, Channel channel);

        default void serverExceptionCaught(ChannelHandlerContext ctx, Throwable e) {
            if (e instanceof IOException) {
                /*
                代码没bug时,一般是client程序非正常结束,没有正常关闭socket
                    导致IO异常:远程主机强迫关闭了一个现有的连接
                    此时onClientConnectChanged已经通知callback了,所以不需要打印
                注意:如果是其他问题,如代码bug引起的IO异常,则此处是一个大坑,会屏蔽异常信息,
                    但代码规范是,实现业务逻辑时,应自己捕获IO异常!
                 */
                return;
            }
            log.error("channel={}", ctx.channel().remoteAddress(), e);
        }

        /**
         * 收到客户端的未知消息
         */
        default void onUnknownData(ChannelHandlerContext ctx, BaseDataPacket data) {
            log.warn("未知数据={}", data);
        }
    }

}
